/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.transport.ptcp.psocket;

import java.io.IOException;
import java.io.OutputStream;

import cgl.narada.transport.ptcp.psocket.threads.PTCPSenderWorker;
import cgl.narada.transport.ptcp.psocket.tools.PTCPDataFragmentator;
import cgl.narada.transport.ptcp.psocket.tools.PTCPPacket;
import cgl.narada.transport.ptcp.psocket.tools.PTCPParameters;

public class PTCPOutputStream extends OutputStream implements PTCPParameters {
  private OutputStream[] output_streams_ = null; //internal list of streams
  private PTCPSenderWorker[] psws_ = null;
  private boolean done_ = false;
  private int number_of_streams_ = 1; //number of streams
  private int packet_number_ = -1;
  private int sender_workers_count_ = 0; //Finished write count
  private String module_name_ = "PTCPOutputStream: ";

  /**
   * Default constructor
   * Construct an empty parallel output stream.
   */

  public PTCPOutputStream(OutputStream[] streams) {
    super();
    int i;
    number_of_streams_ = streams.length;
    output_streams_ = new OutputStream[number_of_streams_];

    psws_ = new PTCPSenderWorker[number_of_streams_];

    Thread worker = null;

    for (i = 0; i < output_streams_.length; i++) {
      output_streams_[i] = streams[i];
      psws_[i] = new PTCPSenderWorker(this, streams[i],
                                   String.valueOf(i));
      worker = new Thread(psws_[i]);
      worker.start();
    }
  }

  /**
   * Closes this output stream and releases any system resources
   * associated with this stream. The general contract of close is that it
   * closes the output stream. A closed stream cannot perform output
   * operations and cannot be reopened.
   */
  public void close() throws IOException {
    if (output_streams_ == null) {
      throw new IOException(module_name_ + "No internal output streams.");
    }

    done_ = true;

    PTCPPacket dummyPacket = new PTCPPacket( -1, new byte[1]);

    for (int i = 0; i < number_of_streams_; i++) {
      psws_[i].wakeup(dummyPacket);
    }

    try {
      for (int i = 0; i < output_streams_.length; i++) {
        output_streams_[i].close();
      }
    }
    catch (IOException e) {
      throw e;
    }
  }

  /**
   * Check whether we are finished or not.
   */
  public synchronized boolean finished() {
    return done_;
  }

  /**
   * Flushes this output stream and forces any buffered output bytes to
   * be written out. The general contract of flush is that calling it is an
   * indication that, if any bytes previously written have been buffered
   * by the implementation of the output stream, such bytes should
   * immediately be written to their intended destination.
   */
  public void flush() throws IOException {
    if (output_streams_ == null) {
      throw new IOException(module_name_ + "No internal output streams.");
    }

    try {
      for (int i = 0; i < output_streams_.length; i++) {
        output_streams_[i].flush();
      }
    }
    catch (IOException e) {
      throw e;
    }
  }

  public void send(byte[] b) throws Exception {
    send(b, 0, b.length);
  }

  public synchronized void send(byte[] b, int off, int len) throws Exception {
    if (b != null) {
      byte[] data = new byte[len - off];
      System.arraycopy(b, off, data, 0, data.length);

      PTCPDataFragmentator df = new PTCPDataFragmentator(data, number_of_streams_,
          packet_number_);
      PTCPPacket[] packets = df.getPackets();
      int packetSize = packets[0].getPacketSize();

      packet_number_ = packets[number_of_streams_ - 1].getPacketNumber();

      for (int i = 0; i < number_of_streams_; i++) {
        if (PTCPOutputStream_Debug) {
          System.out.println(module_name_ + i + " has sent " +
                             packets[i].getPacketNumber() +
                             ". packet " + packets[i].getPacketSize());
        }

        psws_[i].wakeup(packets[i]);
      }

      waitSenderWorkers();
    }
  }

  private int getByteArrayActualLength(byte[] buf) {
    for (int i = (buf.length - 1); i >= 0; i--) {
      if (buf[i] == 0) {
        System.out.println(i + " " + buf[i]);
      }
      else {
        return (i + 1);
      }
    }
    return 0;
  }

  public synchronized void sendEOF() {
    PTCPPacket dummyPacket = new PTCPPacket( -1, new byte[1]);

    for (int i = 0; i < number_of_streams_; i++) {
      psws_[i].wakeup(dummyPacket);
    }

    waitSenderWorkers();
  }

  public synchronized void wakeUp(String name) {
    sender_workers_count_++;
    // if all writers finished writing, wake this stream
    if (sender_workers_count_ == number_of_streams_) {
      notify();
    }
  }

  /**
   * Writes the specified byte to this output stream. The general
   * contract for write is that one byte is written to the output stream.
   * The byte to be written is the eight low-order bits of the argument
   * b. The 24 high-order bits of b are ignored.
   *
   * This single byte will travel on command stream
   */
  public void write(int b) throws IOException {
  }

  /**
   * Writes b.length bytes from the specified byte array to this output
   * stream. The general contract for write(b) is that it should have
   * exactly the same effect as the call write(b, 0, b.length).
   *
   * This is a parallel write.
   */
  public void write(byte[] b) throws IOException {
    try {
      write(b, 0, b.length);
    }
    catch (IOException e) {
      throw e;
    }
  }

  /**
   * Writes len bytes from the specified byte array starting at offset
   * off to this output stream. The general contract for write(b, off,
   * len) is that some of the bytes in the array b are written to the
   * output stream in order; element b[off] is the first byte written
   * and b[off+len-1] is the last byte written by this operation.
   *
   * If b is null, a NullPointerException is thrown.
   * If off is negative, or len is negative, or off+len is greater than the
   * length of the array b, then an IndexOutOfBoundsException is
   * thrown.
   *
   * This is a parallel write.
   *
   * @param b - the data.
   * @param off - the start offset in the data.
   * @param len - the number of bytes to write.
   */
  public void write(byte[] b, int off, int len) throws IOException {
  }

  /**
   * Wait for writers to finish.
   */
  private synchronized void waitSenderWorkers() {
    while (sender_workers_count_ < number_of_streams_) {
      try {
        wait();
      }
      catch (InterruptedException e) {
      }
    }
    sender_workers_count_ = 0;
  }
}
